package com.ahaha.Actions

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.hadoop.io.{IntWritable, Text}
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat

object Actions {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf()
            .setAppName("ActionOps")
            .setMaster("local[*]")
        val sc = new SparkContext(conf)

        val lines = sc.parallelize(List(
            "hello you",
            "hello me",
            "hello lan lan"
        ))

        val pairs = lines.flatMap(_.split("\\s+")).map((_, 1))
        var ret: RDD[(String, Int)] = pairs.reduceByKey(_ + _)
        ret.foreach(println)
        println("=============countByKey")
        val cbk = pairs.countByKey()
        println(cbk)

        println("=============countByValue")
        val cbv = pairs.countByValue()
        cbv.foreach{ case ((key, value),longval) => {
            println(s"key:${key}---value:${value}------longval:${longval}")
        }}


        println("========take")
        var listRDD = sc.parallelize(1 to 10, 2)
        listRDD = listRDD.mapPartitionsWithIndex((index, partition) => {
            val list = partition.toList
            println(s"rdd中partition<${index}>中的数据为：" + list.mkString("[", ", ", "]"))
            list.toIterator
        })

        val takearray:Array[Int] = listRDD.take(8)

        takearray.foreach(println)

        println("=========first")
        val firstvalue = listRDD.first()
        println(firstvalue)

        println("===========collect")
        val arr = listRDD.collect()
        arr.foreach(println)

        println("=========reduce")

        val sum = listRDD.reduce(_ + _)
        println("sum:" + sum)

        println("=========save")
        //var ret:RDD[(String, Int)] = pairs.reduceByKey(_+_)

        ret.saveAsNewAPIHadoopFile(
            path = "hdfs://hadoop003:9000/tmp/mytmp/save",
            keyClass = classOf[Text],
            valueClass = classOf[IntWritable],
            outputFormatClass = classOf[TextOutputFormat[Text,IntWritable]]
        )

        sc.stop()
    }
}
